package com.dxy.chatserver.subscription;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import com.aliyun.openservices.ons.shaded.commons.lang3.StringUtils;
import com.dxy.chatserver.builder.RocketBuilder;
import com.dxy.chatserver.config.Event;
import com.dxy.chatserver.db.entity.AcceptLineMsgResponse;
import com.dxy.chatserver.db.entity.CleanMsg;
import com.dxy.chatserver.db.entity.MultiMessage;
import com.dxy.chatserver.service.EsService;
import lombok.extern.slf4j.Slf4j;
import org.bson.types.ObjectId;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

/**
 * 批量消息监听
 *
 * @author: none
 * @date: 2021/12/26
 **/
@Service
@Slf4j
public class MsgBatchListener implements BatchMessageListener {

    @Resource
    private EsService esService;

    /**
     * 批量消费消息接口，由应用来实现<br>
     * 需要注意网络抖动等不稳定的情形可能会带来消息重复，对重复消息敏感的业务可对消息做幂等处理
     *
     * @param messages 一批消息
     * @param context  消费上下文
     * @return {@link Action} 消费结果，如果应用抛出异常或者返回Null等价于返回Action.ReconsumeLater
     * @see <a href="https://help.aliyun.com/document_detail/44397.html">如何做到消费幂等</a>
     *
     *
     * 项目特殊原因 这个只处理ES
     */
    @Override
    public Action consume(List<Message> messages, ConsumeContext context) {
        try {
            Map<String, List<AcceptLineMsgResponse>> map = new LinkedHashMap<>();
            Map<String, List<CleanMsg>> cleanMsg = new LinkedHashMap<>();
            for (Message message : messages) {
                MultiMessage<?> msg = RocketBuilder.MAPPER.readValue(message.getBody(), MultiMessage.class);
                if (Objects.nonNull(msg)) {
                    if (StringUtils.isNotBlank(msg.getSession())) {
                        if (StringUtils.equals(Event.clearAllMsg, msg.getEvent())) {
                            // 清除通知
                            List<CleanMsg> responses = cleanMsg.get(msg.getSession());
                            if (Objects.isNull(responses)) {
                                responses = new LinkedList<>();
                                cleanMsg.put(msg.getSession(), responses);
                            }
                            CleanMsg clean = RocketBuilder.MAPPER.convertValue(msg.getResponse(), CleanMsg.class);
                            clean.setId(new ObjectId(clean.getMsgId()));
                            responses.add(clean);
                        } else {
                            // 聊天消息
                            List<AcceptLineMsgResponse> responses = map.get(msg.getSession());
                            if (Objects.isNull(responses)) {
                                responses = new LinkedList<>();
                                map.put(msg.getSession(), responses);
                            }
                            log.debug("解析消息: {}", msg);
                            responses.add(RocketBuilder.MAPPER.convertValue(msg.getResponse(), AcceptLineMsgResponse.class));
                        }
                    } else if (msg.getType() == 6) {
                        // socket连接执行操作
/*                        group.execute(() -> {
                            SocketClient client = RocketBuilder.MAPPER.convertValue(msg.getResponse(), SocketClient.class);
                            // 发送删除通知
                            connectManager.sendCleanMsg(client);
                            // 发送群组通知
                            connectManager.sendNoticeMsg(client);
                        });*/
                    } else if (msg.getType() == 7) {
                        // socket断开
                    }
                }
            }
            // 保存
            this.save(map, cleanMsg);
        } catch (Exception e) {
            log.error("保存聊天数据异常!", e);
            return Action.ReconsumeLater;
        }
        return Action.CommitMessage;
    }

    /**
     * 保存消息
     *
     * @param onlineMsg 在线消息
     * @param cleanMsg  清除记录消息
     */
    private void save(Map<String, List<AcceptLineMsgResponse>> onlineMsg, Map<String, List<CleanMsg>> cleanMsg) {
        onlineMsg.forEach((k, v) -> {
            // 保存消息
            esService.addBatch(v,k);
        });
    }
}
